package geektime.spring.data.reactive.r2dbcrepositorydemo;

import geektime.spring.data.reactive.r2dbcrepositorydemo.converter.MoneyReadCoverter;
import geektime.spring.data.reactive.r2dbcrepositorydemo.converter.MoneyWriteConverter;
import geektime.spring.data.reactive.r2dbcrepositorydemo.repository.CoffeeRepository;
import io.r2dbc.h2.H2ConnectionConfiguration;
import io.r2dbc.h2.H2ConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.convert.CustomConversions.StoreConversions;
import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
import reactor.core.publisher.Flux;

@EnableR2dbcRepositories
@Slf4j
@SpringBootApplication
public class R2dbcRepositoryDemoApplication extends AbstractR2dbcConfiguration
		implements ApplicationRunner {

	@Autowired
	private CoffeeRepository coffeeRepository;

	public static void main(String[] args) {
		SpringApplication.run(R2dbcRepositoryDemoApplication.class, args);
	}


	@Bean
	public ConnectionFactory connectionFactory() {
		return new H2ConnectionFactory(
				H2ConnectionConfiguration.builder().inMemory("testdb").username("sa").build());
	}

	@Bean
	public R2dbcCustomConversions r2dbcCustomConversions() {
		Dialect dialect = getDialect(connectionFactory());
		CustomConversions.StoreConversions storeConversions = StoreConversions
				.of(dialect.getSimpleTypeHolder());
		return new R2dbcCustomConversions(storeConversions,
				Arrays.asList(new MoneyReadCoverter(), new MoneyWriteConverter()));
	}

	@Override
	public void run(ApplicationArguments args) throws Exception {
		CountDownLatch cdl=new CountDownLatch(2);

		coffeeRepository.findAllById(Flux.just(1L,2L))
				.map(c->c.getName()+"-"+c.getPrice().toString())
				.doFinally(s->cdl.countDown())
				.subscribe(c->log.info("Find {}",c));

		coffeeRepository.findByName("mocha")
				.doFinally(s->cdl.countDown())
				.subscribe(c->log.info("Find {}",c));

		cdl.await();
	}
}
